package com.leo.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

@Configuration
@EnableRabbit
public class RabbitConfig {

    @Bean(name = "rabbitMessageListenerContainer")
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory) {
        //写的时候，默认使用DEFAULT_NUM_THREADS =Runtime.getRuntime().availableProcessors() * 2个线程
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(5);

        // 并发由setQueues和setConsumersPerQueue控制
        // 从哪些队列消费消息，就设置哪些队列
//        container.setQueues(orderQueue(), ttlQueue(), dlxQueue());
//        container.setQueueNames("", "", "", "");
        // 设置每个队列由几个消费者消费。
        // 如果容器已经在运行，该值可以动态设置。
        // 每个消费者使用一个Channel

        /**
         * 每个队列的消费者数量
         */
        container.setConsumersPerQueue(5);
        // 一次确认几个消息
        container.setMessagesPerAck(1);

        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(20);
        // 设置该属性，灵活设置并发
        container.setTaskExecutor(taskExecutor);

        return container;
    }

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 订单消息队列
     * @return
     */
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("q.order").build();
    }

    /**
     * 延迟消息队列
     * @return
     */
    @Bean
    public Queue ttlQueue() {
        Map<String, Object> args = new HashMap<>();
        // 10s消息到期
        args.put("x-message-ttl", 10000);
        args.put("x-dead-letter-exchange", "x.dlx");
        args.put("x-dead-letter-routing-key", "k.dlx");
        Queue queue = new Queue("q.ttl", true, false, false, args);

        return queue;
    }

    /**
     * 死信队列，用于取消用户订单
     * @return
     */
    @Bean
    public Queue dlxQueue() {
        Map<String, Object> args = new HashMap<>();
        Queue dlq = new Queue("q.dlx", true, false, false, args);

        return dlq;
    }

    /**
     * 订单交换器
     * @return
     */
    @Bean
    public Exchange orderExchange() {
        Map<String, Object> args = new HashMap<>();
        DirectExchange exchange = new DirectExchange("x.order", true, false, args);

        return exchange;
    }

    /**
     * 延迟队列交换器
     * @return
     */
    @Bean
    public Exchange ttlExchange() {
        Map<String, Object> args = new HashMap<>();
        return new DirectExchange("x.ttl", true, false, args);
    }

    /**
     * 死信队列交换器
     * @return
     */
    @Bean
    public Exchange dlxExchange() {
        Map<String, Object> args = new HashMap<>();
        DirectExchange exchange = new DirectExchange("x.dlx", true, false, args);
        return exchange;
    }

    /**
     * 用于发送下单，做分布式事务的MQ
     * @return
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("k.order").noargs();
    }

    /**
     * 用于等待用户支付的延迟队列绑定
     * @return
     */
    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("k.ttl").noargs();
    }

    /**
     * 用于支付超时取消用户订单的死信队列绑定
     * @return
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("k.dlx").noargs();
    }

}
